package com.theaf.core.rmq;

import com.theaf.core.rmq.common.Constants;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by littlersmall on 16/5/23.
 */
@Slf4j
public class ThreadPoolConsumer<T> {
    private ExecutorService executor;
    private volatile boolean stop = false;
    private int threadCount;

    private static List<MessageConsumer> messageConsumerList = Collections.synchronizedList(new ArrayList<MessageConsumer>());
    public static void addConsumer(MessageConsumer messageConsumer){
        messageConsumerList.add(messageConsumer);
    }

    private static ThreadPoolConsumer instance;

    public static void instance(int threadCount){
        if(instance!=null) {
            instance = new ThreadPoolConsumer(threadCount);
            instance.start();
        }
    }

    private ThreadPoolConsumer(int threadCount ) {
        this.threadCount = threadCount;
        executor = Executors.newFixedThreadPool(threadCount);
    }

    //1 构造messageConsumer
    //2 执行consume
    public void start() {
        for (int i=0;i<threadCount;i++) {
            final int ii = i;
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    while (!stop) {
                        for (int j = 0; j < messageConsumerList.size(); j++)
                            if (j % threadCount == ii) {
                                try {
                                    messageConsumerList.get(j).consume();
                                    try {
                                        Thread.sleep(60 * 1000);
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                        log.info("interrupt ", e);
                                    }
                                } catch (Exception e) {
                                    e.printStackTrace();
                                    log.info("run exception ", e);
                                }
                            }
                    }
                    try {
                        Thread.sleep(60*1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        log.info("interrupt ", e);
                    }
                }
            });
        }

        Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
    }

    public void stop() {
        this.stop = true;
        try {
            Thread.sleep(Constants.ONE_SECOND);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
